package luluouter.msg.inner;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Inet4Address;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import com.alibaba.fastjson.JSON;

import luluouter.data.inner.InnerDataMaster;
import luluouter.msg.model.Mole;
import luluouter.msg.model.Pigeon;

public class InnerMsgClient implements Runnable {
    private final static int FLAG_MSG_HELLO = 2017;
    private final static int FLAG_MSG_CREATE_DATA_PIPE = 2048;
    private final static int FLAG_MSG_HEART_BEAT = 1986;
    private final static String MSG_KEY_HEART_BEAT = "heartBeat";

    private BlockingQueue<Map<String, Object>> outQueue = new LinkedBlockingQueue<Map<String, Object>>();

    private Pigeon pigeon;

    private final Socket inner;

    private volatile boolean stop = true;

    public InnerMsgClient(Socket inner) {
        this.inner = inner;
        stop = false;
        System.out.println("InnerMsgClient:" + inner.getRemoteSocketAddress());
    }

    @Override
    public void run() {
        try (Socket closeSocket = inner;
                OutputStream outputStream = inner.getOutputStream();
                DataOutputStream out = new DataOutputStream(outputStream);
                InputStream inputStream = inner.getInputStream();
                DataInputStream in = new DataInputStream(inputStream);) {
            while (!stop) {
                Map<String, Object> msg = outQueue.take();
                if (!msg.containsKey(MSG_KEY_HEART_BEAT)) {
                    System.out.println("InnerMsgClient.take:" + msg);
                }

                String json = JSON.toJSONString(msg);
                out.writeUTF(json);
                out.flush();
                if (!msg.containsKey(MSG_KEY_HEART_BEAT)) {
                    System.out.println("InnerMsgClient.writeUTF:" + json);
                }

                String line = in.readUTF();
                if (!line.contains(MSG_KEY_HEART_BEAT)) {
                    System.out.println("InnerMsgClient.readUTF:" + line);
                }
                Map<String, Object> result = JSON.parseObject(line, Map.class);
                dealMsg(result);
            }
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } finally {
            InnerMsgClientMaster.getInstance().removeInnerMsgClient(pigeon);
            System.out.println("InnerMsgClient.finally:" + pigeon);
        }
    }

    public void init() {
        Map<String, Object> msg = new HashMap<String, Object>();
        msg.put("msgType", FLAG_MSG_HELLO);
        addMsg(msg);
        System.out.println("InnerMsgClient.init:" + msg);
    }

    public void addMsg(Map<String, Object> msg) {
        boolean result = outQueue.offer(msg);
        System.out.println("InnerMsgClient.outQueue.offer:" + result + "/" + msg);
    }

    private void dealMsg(Map<String, Object> msg) {
        Integer msgType = (Integer) msg.get("msgType");
        if (msgType == FLAG_MSG_HELLO) {
            // proxyPort = (Integer) msg.get("proxyPort");

            SocketAddress socketAddress = inner.getRemoteSocketAddress();
            String address = String.valueOf(socketAddress);
            if (socketAddress instanceof InetSocketAddress) {
                InetAddress inetAddress = ((InetSocketAddress) socketAddress).getAddress();
                if (inetAddress instanceof Inet4Address) {
                    address = String.valueOf(((Inet4Address) inetAddress).getHostAddress());
                }
            }

            String code = String.valueOf(msg.get("code"));
            pigeon = new Pigeon(address, code);

            InnerMsgClientMaster.getInstance().addInnerMsgClient(pigeon, this);
        } else if (msgType == FLAG_MSG_CREATE_DATA_PIPE) {
            Object indexObj = msg.get("index");
            String indexStr = String.valueOf(indexObj);
            Long index = Long.parseLong(indexStr);

            String result = (String) msg.get("result");
            if ("success".equals(result)) {
                InnerDataMaster.getInstance().startPipes(index);
            } else {
                InnerDataMaster.getInstance().destroyOuterClient(index);
            }
        }
    }

    public void createPipes(Socket outerClient, Mole mole) {
        long theIndex = InnerDataMaster.getInstance().addOuterClient(outerClient);

        Map<String, Object> msg = new HashMap<String, Object>();
        msg.put("msgType", FLAG_MSG_CREATE_DATA_PIPE);
        msg.put("index", theIndex);

        msg.put("inner_ip", mole.getIp());
        msg.put("inner_port", mole.getPort());

        addMsg(msg);
    }

    public void heartBeat() {
        long ts = System.currentTimeMillis();
        Map<String, Object> msg = new HashMap<String, Object>();
        msg.put("msgType", FLAG_MSG_HEART_BEAT);
        msg.put("date", new Date(ts));
        msg.put(MSG_KEY_HEART_BEAT, ts);
        outQueue.offer(msg);
        // System.out.println("InnerMsgClient.heartBeat:" + msg);
    }
}
